package com.uni.hbase.utils;

/*
 * @Author: Uni
 * @Time: 2022/5/3
 * @TODO 实现 HBase 的 DDL 和 DML 操作
 */

import org.apache.commons.lang.NullArgumentException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
import org.apache.hadoop.hbase.util.Bytes;

import javax.validation.constraints.Null;
import java.io.IOException;

/**
 * DDL:
 *      1. 判断表是否存在
 *      2. 创建表
 *      3. 创建命名空间
 *      4. 删除表
 *  <p>
 *  DML:
 *      5. 插入数据
 *      6. 查询数据（get）
 *      7  查询数据（scan）
 *      8. 删除数据
 */
public class TestAPI {

    private static Connection connection = null;
    private static Admin admin = null;

    static {
        try{
            // 1. 获取配置信息
            Configuration config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum","hadoop101,hadoop102,hadoop103");
            // 2. 创建连接对象
            connection = ConnectionFactory.createConnection(config);
            // 3. 创建 Admin 对象
            admin = connection.getAdmin();
        } catch (IOException e){
            e.printStackTrace();
        }
    }

    /**
     * 判断 HBase 表 是否存在
     * @param tableName
     * @return
     * @throws IOException
     */
    public static boolean isTableExist(String tableName) {
        boolean exists = false;
        try {
            exists = admin.tableExists(TableName.valueOf(tableName));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return exists;
    }


    /**
     * 创建表
     * @param tableName
     * @param cfs
     * @throws TableExistsException
     */
    public static void createTable(String tableName, String... cfs)
            throws TableExistsException {
        // 1. 判断是否存在列族信息
        if(cfs.length <= 0)
            throw new NullArgumentException("缺少列族信息");
        // 2. 判断表是否存在
        if(isTableExist(tableName))
            throw new TableExistsException("表已存在");
        // 3. 创建表描述器
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
        // 4. 循环添加列族信息
        for (String cf : cfs) {
            // 5. 创建列族描述器
            HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cf);

            // 6. 添加具体的列族信息
            hTableDescriptor.addFamily(hColumnDescriptor);
        }
        // 创建表
        try {
            admin.createTable(hTableDescriptor);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除表
     */
    public static void dropTable(String tableName) throws TableNotFoundException {
        // 1. 判断表是否存在
        if(!isTableExist(tableName))
            throw new TableNotFoundException("表不存在");
        try {
            // 2. 使表下线
            admin.disableTable(TableName.valueOf(tableName));
            // 3. 删除表
            admin.deleteTable(TableName.valueOf(tableName));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建 HBase表的域名空间
     * @param ns
     */
    public static void createNameSpace (String ns){
        // 1. 创建命名空间描述器
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(ns).build();
        // 2. 创建命名空间
        try {
            admin.createNamespace(namespaceDescriptor);
        } catch (NamespaceExistException e){
            System.out.println(ns + "命名空间已存在!");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * 向表插入数据
     */
    public static void putData(String tableName, String rowKey, String cf, String cn, String value){
        try {
            // 1. 获取表对象
            Table table = connection.getTable(TableName.valueOf(tableName));
            // 2. 创建 Put 对象
            Put put = new Put(Bytes.toBytes(rowKey));
            // 3. 给 Put对象赋值
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn), Bytes.toBytes(value));
            // 4. 插入数据
            table.put(put);
            // 5. 关闭表连接
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取数据
     */
    public static void getData(String tableName, String rowKey, String cf, String cn){
        try {
            // 1. 获取表对象
            Table table = connection.getTable(TableName.valueOf(tableName));
            // 2. 创建get对象
            Get get = new Get(Bytes.toBytes(rowKey));
            //  2.1 指定获取的列族
//            get.addFamily(Bytes.toBytes(cf));
            //  2.2  指定列族和列
            get.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cf));

            // 2.3 设置获取数据的版本数 (设置获取最大)
            int maxVersions = get.readAllVersions().getMaxVersions();
            get.readVersions(maxVersions);
            // 3. 获取数据
            Result result = table.get(get);
            // 4. 解析 result 并打印
            for (Cell cell : result.rawCells()) {
                // 5. 打印数据
                String queryCf = Bytes.toString(CellUtil.cloneFamily(cell));
                String queryCn = Bytes.toString(CellUtil.cloneQualifier(cell));
                String val = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("cf:" + queryCf
                            + ", cn:" + queryCn
                            + ", val:" + val);
            }
            // 6. 关闭表连接
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }


    }
    /**
     * 关闭资源
     */
    public static void close(){
        if(admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if(connection!=null){
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) throws IOException {
        // 测试表是否存在（默认namespace，若有则为namespace:tableName
        System.out.println(isTableExist("stu"));

        // 创建表测试
        createTable("stu5", "info1", "info2");
        // 删除表测试
        dropTable(("stu5"));
        System.out.println(isTableExist("stu5"));
        // 创建命名空间测试: (创建后可在shell里输入list_namespace查询)
        createNameSpace("521");

        // 创建数据测试
        putData("stu", "1001", "info",
                "name", "chao");

        // 获取单行数据
        getData("stu", "1001", "info", "name");
        // 关闭资源
        close();
    }
}
